package SparkExercises.kejian4;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.List;

public class LianXi4 {
    public static void main(String[] args) {
        /**
         * 给定两份数据: （本市违章记录）和（临市B违章记录），请写一个spark程序，
         * 查找出在本市以及临市B总共违章次数超过三次以上的车辆。
         * 说明：两份数据结构相同（日期 设备编号 车牌号 违章类型码），列与列之间，均已tab作为分隔符.
         * 思路：
         * 1）用textFile算子分别读取两个文件，得到两个RDD对象
         * 2）union合并两个RDD
         * 3）用mapToPair算子将上一步得到的合并后的RDD转成键值对RDD，key是车牌，value是1
         * 4）再用reduceByKey计算每个车牌的累积违章次数，得到一个新的键值对RDD
         * 5）基于上一步得到的键值对RDD，用filter算子过滤出违章次数大于三次的元素
         * 将filter得到的RDD元素打印出来
         */
        SparkConf sparkConf = new SparkConf().setAppName("LianXi4").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        JavaRDD<String> rdd1 = sc.textFile("./data/records.txt");
        JavaRDD<String> rdd2 = sc.textFile("./data/recordsCityB.txt");

        JavaRDD<String> rdd3 = rdd1.union(rdd2);

        JavaPairRDD<String, Integer> rdd4 = rdd3.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                String[] split = s.split("\t");
                return new Tuple2<>(split[2], 1);
            }
        });

        JavaPairRDD<String, Integer> rdd5 = rdd4.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        });
        // 键值对 key _1 value _2
        JavaPairRDD<String, Integer> rdd6 = rdd5.filter(new Function<Tuple2<String, Integer>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                if (stringIntegerTuple2._2 > 3) {
                    return true;
                } else {
                    return false;
                }

            }
        });


        List<Tuple2<String, Integer>> collect = rdd6.collect();
        for (Tuple2<String, Integer> t : collect){
            System.out.println(t);
        }

//        List<String> collect = rdd3.collect();
//        for (String s : collect){
//            System.out.println(s);
//        }


    }
}
